﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

namespace MT4Trainer.Parallel_Helpers
{
    using System.Collections.Concurrent;
    using System.Threading.Tasks;

    public class Pipeline<TInput, TOutput>
    {
        // queue based blocking collection
        private BlockingCollection<ValueCallBackWrapper> valueQueue;

        // the function to use
        private Func<TInput, TOutput> pipelineFunction;

        public Pipeline(Func<TInput, TOutput> function)
        {
            // assign the function
            pipelineFunction = function;
        }

        public Pipeline<TInput, TNewOutput> AddFunction<TNewOutput>(Func<TOutput, TNewOutput> newfunction)
        {
            // create a composite function
            Func<TInput, TNewOutput> compositeFunction = (inputValue =>
                { return newfunction(pipelineFunction(inputValue)); });
            // return a new pipeline around the composite function
            return new Pipeline<TInput, TNewOutput>(compositeFunction);
        }

        public void AddValue(TInput value, Action<TInput, TOutput> callback)
        {
            // add the value to the queue for processing
            valueQueue.Add(new ValueCallBackWrapper { Value = value, Callback = callback });
        }

        public void StartProcessing()
        {
            // initialize the collection
            valueQueue = new BlockingCollection<ValueCallBackWrapper>();
            // create a parallel loop to consume
            // items from the collection
            Task.Factory.StartNew(
                () =>
                    {
                        Parallel.ForEach(
                            valueQueue.GetConsumingEnumerable(),
                            wrapper => wrapper.Callback(wrapper.Value, this.pipelineFunction(wrapper.Value)));
                    });
        }

        public void StopProcessing()
        {
            // signal to the collection that no
            // further values will be added
            valueQueue.CompleteAdding();
        }

        private class ValueCallBackWrapper
        {
            public TInput Value;

            public Action<TInput, TOutput> Callback;
        }


        public void PipeLineExample()
        {
            // create a set of functions that we want to pipleline together
            Func<int, double> func1 = (input => Math.Pow(input, 2));
            Func<double, double> func2 = (input => input / 2);
            Func<double, bool> func3 = (input => input % 2 == 0 && input > 100);

            Func<double, double> powered = (input => Math.Pow(input, 2));
           

            

            // define a callback
            Action<int, bool> callback = (input, output) =>
                {
                    if (output)
                    {
                        Console.WriteLine("Found value {0} with result {1}", input, output);
                    }
                };
            // create the pipeline
            Pipeline<int, bool> pipe = new Pipeline<int, double>(func1).AddFunction(func2).AddFunction(func3);
            // start the pipeline
            pipe.StartProcessing();
            // generate values and push them into the pipeline
            for (int i = 0; i < 1000; i++)
            {
                Console.WriteLine("Added value {0}", i);
                pipe.AddValue(i, callback);
            }
            // stop the pipeline
            pipe.StopProcessing();
            // wait for input before exiting
            Console.WriteLine("Press enter to finish");
            Console.ReadLine();
        }
    }
}